package com.hba.broker;

import com.hba.constant.RabbitConstant;
import com.hba.init.RabbitInitConfigure;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

@Component
public class RabbitBroker {
    @Autowired
    AmqpAdmin amqpAdmin;
    @Autowired
    RabbitInitConfigure rabbitInitConfigure;

    private static Set<String> directBindCacheSet = new CopyOnWriteArraySet<String>();
    private static ConcurrentHashMap<String,DirectExchange> directExchangeMap = new ConcurrentHashMap<String, DirectExchange>();
    private static ConcurrentHashMap<String,Queue> queueMap = new ConcurrentHashMap<String, Queue>();

    public static int num = 10;

    public void declareBindExchange(String exchangeName,String queueName,String toExchange,String routingKey){
        if(directBindCacheSet.contains(exchangeName+"|"+queueName)){
            return;
        }
        DirectExchange directExchange = null;
        if(!directExchangeMap.contains(exchangeName)){
            directExchange =  new DirectExchange(exchangeName,true,false,null);
            amqpAdmin.declareExchange(directExchange);
            directExchangeMap.putIfAbsent(exchangeName,directExchange);
        }
        Queue queue = null;
        if(!queueMap.contains(queueName)){
            Map<String,Object> map = new HashMap<String,Object>(2,1F);
            //map.put("x-dead-letter-exchange", RabbitConstant.DEFAULT_DEAD_LETTER_EXCHANGE);
            //map.put("x-dead-letter-routing-key", queueName+RabbitConstant.DEFAULT_SEQUEUE_RETRY_SUFFIX);
            map.put("x-dead-letter-exchange", toExchange);
            map.put("x-dead-letter-routing-key", routingKey);
            for(int i = 0; i< num; i++){
                String logicQueue = queueName + "_" + i;
                Queue queueI = new Queue(logicQueue,true,false,false ,map);
                queue = queueI;
                amqpAdmin.declareQueue(queueI);
                String bindRelation = exchangeName+"|"+queueName;
                if(directBindCacheSet.contains(bindRelation)){
                    continue;
                }
                amqpAdmin.declareBinding(BindingBuilder.bind(queueI).to(directExchange).with(logicQueue));
            }
            queueMap.putIfAbsent(queueName,queue);
            directBindCacheSet.add(exchangeName+"|"+queueName);
        }
    }

    public void declareDeadLetterBindExchange(String exchangeName,String queueName,String toExchangeName,String routingKey){
        if(directBindCacheSet.contains(exchangeName+"|"+queueName)){
            return;
        }
        DirectExchange directExchange = null;
        if(!directExchangeMap.contains(exchangeName)){
            directExchange =  new DirectExchange(exchangeName,true,false,null);
            amqpAdmin.declareExchange(directExchange);
            directExchangeMap.putIfAbsent(exchangeName,directExchange);
        }
        Queue queue = null;
        if(!queueMap.contains(queueName)){
            Map<String,Object> map = new HashMap<String,Object>(2,1F);
            map.put("x-dead-letter-exchange", toExchangeName);
            map.put("x-dead-letter-routing-key", routingKey);
            int retryTime = RabbitConstant.DEFAULT_WAIT_TO_RETRY_TIME;
            String retryTimeStr=String.valueOf(rabbitInitConfigure.getRetryTime());
            if(!org.springframework.util.StringUtils.isEmpty(retryTimeStr)){
                retryTime=Integer.valueOf(String.valueOf(retryTimeStr));
            }
            map.put("x-message-ttl", retryTime);
            queue = new Queue(queueName,true,false,false ,map);
            amqpAdmin.declareQueue(queue);
            queueMap.putIfAbsent(queueName,queue);
        }
        String bindRelation = exchangeName+"|"+queueName;
        if(directBindCacheSet.contains(bindRelation)){
            return;
        }
        amqpAdmin.declareBinding(BindingBuilder.bind(queue).to(directExchange).with(queueName));
        directBindCacheSet.add(bindRelation);
    }

}
